package com.yangdage.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

@Service
public class MQServiceImpl  implements  MQService{
    private Logger logger = LoggerFactory.getLogger(MQServiceImpl.class);
    private static final String MESSAGE_KEY = "message:queue";
    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void produce(String string) {
        redisTemplate.opsForList().leftPush(MESSAGE_KEY, string);
    }

    @Override
    public void consume() {
        String string = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY);
        logger.info("consume : {}", string);
    }


    /***
     * 阻塞队列
     */
    public void blockingConsume() {
        List<Object> obj = redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Nullable
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                //队列没有元素会阻塞操作，直到队列获取新的元素或超时
                return connection.bLPop(20, MESSAGE_KEY.getBytes());
            }
        },new StringRedisSerializer());

        for (Object str: obj) {
            logger.info("blockingConsume : {}", str);
        }
    }
}
